#include "MessageHistoryService_types.h"
#include "MessageHistoryService.h"

#include <boost/scoped_ptr.hpp>
#include <boost/shared_ptr.hpp>
//#include <boost/detail/atomic_count.hpp>
#include <boost/foreach.hpp>

#include "message_redis_adapter.h"
#include "message_mysql_adapter.h"
#include "call_center_cache.h"
#include "base/logging.h"
#include "base/get_local_ip.h"
#include "base/config_reader.h"
#include "configer.h"
#include "melon/service/service_arch.h"
#include "melon/service/base_service_impl.h"
#include "redis_executor/simple_redis_locator.h"

#include "im/adapters/im_call_center_adapter.h"

#include "thread_manager.h"

using boost::scoped_ptr;
using boost::shared_ptr;
//using boost::detail::atomic_count;
using apache::thrift::concurrency::Runnable;

namespace imserver {

class AddMsgTask : public Runnable {
public:
  AddMsgTask(MessageRedisAdapter * redis_adapter, MessageMysqlAdapter * mysql_adapter, const ImMessage& message) 
      : redis_adapter_(redis_adapter), mysql_adapter_(mysql_adapter), message_(message) {
  }

  virtual void run() {
    for(int i = 0; i < 3; ++i) {
      if (0 == redis_adapter_->AddImMessage(message_)) {
        LOG_INFO("redis AddMessage ok, id=" << message_.seq_id << " sender=" << message_.sender
             << " receiver=" << message_.receiver);
        break;
      }
      LOG_WARN("redis AddMessage " << message_.seq_id << " err, sender=" << message_.sender
             << " receiver=" << message_.receiver);
    }
    int res = mysql_adapter_->InsertImMessage(message_);
    LOG_INFO("redis AddMessage " << message_.seq_id << " res=" << res << " sender=" << message_.sender
        << " receiver=" << message_.receiver);
  }
private:
  MessageRedisAdapter * redis_adapter_;
  MessageMysqlAdapter * mysql_adapter_;
  ImMessage message_;
};
/*
class UpdateMsgTask : public Runnable {
public:
  UpdateMsgTask(shared_ptr<MessageMysqlAdapter> obj, const std::vector<int64_t> & seq_ids, const int64_t uid) : runner_(obj), ids_(seq_ids), uid_(uid) {}
  virtual void run() {
    runner_->UpdatereadFlagByID(ids_, uid_);
  }
private:
  shared_ptr<MessageMysqlAdapter> runner_;
  std::vector<int64_t> ids_;
  int64_t uid_;
};
*/
class MessageHistoryServiceHandler : public MessageHistoryServiceIf, public melon::service::BaseServiceImpl {
public:
  MessageHistoryServiceHandler() : /*seq_id_step_(1000), seq_id_conf_("./seq_id.conf"), index_(NULL), */redis_(NULL) {
    ConfigReader config("../conf/message_history_service.conf");
    redis_.reset(new MessageRedisAdapter(new FixedLocator(config.Get("redis", "server_list").c_str())));
    mysql_.reset(new MessageMysqlAdapter(
          config.Get("database", "name"), config.Get("database", "server_list"),
          config.Get("database", "user"), config.Get("database", "password"))
        );
  }

  virtual ~MessageHistoryServiceHandler() {}

  virtual void AddMessage(const ImMessage& message) {
  /*  if (!index_ || ((*index_) % seq_id_step_ == 0)) {
      int64_t id = seq_id_conf_.reset("id", seq_id_step_);
      LOG_INFO("read id from file is " << id);
      index_.reset(new atomic_count(id));
    }

    const_cast<ImMessage&>(message).__set_seq_id(++(*index_));*/
    if (message.seq_id <= 0) {
      LOG_ERROR("the message(" << message.body << ") with an invalid sequence id(" << message.seq_id 
          << ") sender - receiver " << message.sender << " - " << message.receiver);
      return;
    }

    GetThreadManager()->add(boost::shared_ptr<Runnable>(new AddMsgTask(redis_.get(), mysql_.get(), message)));
  }

  virtual void RemoveMessage(const int64_t sender, const int64_t receiver, const int64_t seq_id) {
    LOG_INFO("RemoveMessage called.");
    redis_->RemoveImMessage(sender, receiver, seq_id);
  }

  virtual void SetRead(const int64_t sender, const int64_t receiver) {
    LOG_INFO("SetRead called.");
  }

  virtual void SetReadByID(const std::vector<int64_t> & seq_ids, const int64_t receiver) {
    LOG_INFO("SetReadByID called. receiver=" << receiver);
    std::vector<int64_t> shops = CallCenterCache::Instance().GetWaiterShops(receiver);
    if (shops.size() > 0) {
      for(size_t i = 0; i < shops.size(); ++i) {
        redis_->SetReadImMessageByID(seq_ids, shops[i]);
      }
    } else {
      redis_->SetReadImMessageByID(seq_ids, receiver);
    }
    //GetThreadManager()->add(boost::shared_ptr<Runnable>(new UpdateMsgTask(mysql_, seq_ids, receiver)));
  }

  virtual void GetOfflineMessages(MessageResult& _return, const int64_t receiver, const int32_t offset, const int32_t limit) {
    /*int ret = redis_->GetOfflineImMessages(_return, receiver, offset, limit);
    int size = _return.results.size();
    if (size < limit && ret == 0) {
      ImMessageList cell;
      int32_t cnt = limit - size + 1; //多取一条，检查是否有更多
      mysql_->SelectUnreadMessages(cell, receiver, offset + size, cnt);
      _return.count = (cell.size() < cnt) ? 0 : -1;//如果取到的数据不够需要取的最大限制，说明没有更多了0表示没有更多

      BOOST_FOREACH(const ImMessage& msg, cell) {
        if (_return.results.size() < limit)
          _return.results.push_back(msg);
      }

      if (_return.count == 0)
        redis_->FlagUnread(receiver);
    }*/
    std::vector<int64_t> shops = CallCenterCache::Instance().GetWaiterShops(receiver);
    if (shops.size() > 0) {
      _return.count = 0;
      for(size_t i = 0; i < shops.size(); ++i) {
        MessageResult r;
        for(int j = 0; j < 2; ++j) {
          if (redis_->GetOfflineImMessages(r, shops[i], offset, limit) >= 0) {
            break;
          }
        }
        LOG_INFO("GetOfflineMessages waiter=" << receiver << " shop=" << shops[i] << " size=" << r.results.size());
        if (r.count > 0 || r.results.size() > 0) {
          _return.results.insert(_return.results.end(), r.results.begin(), r.results.end());
          // _return.results.push_back(r.results[i]);
          _return.count += r.count;
        }
      }
      LOG_INFO("GetOfflineMessages waiter=" << receiver << " total_count=" << _return.count << " total_size=" << _return.results.size());
      return;
    }

    for(int i = 0; i < 2; ++i) {
      if (redis_->GetOfflineImMessages(_return, receiver, offset, limit) >= 0) {
        break;
      }
    }
    LOG_INFO("GetOfflineMessages  receiver=" << receiver << " ret_len=" << _return.results.size());
  }

  virtual void GetMessages(MessageResult& _return, const int64_t sender, const int64_t receiver, const int32_t offset, const int32_t limit) {
    // TODO : 如果sender是 call center waiter, 则需要将其转换为相应的receiver的最近联系人列表中，找到对应的shop
    int64_t effective_sender = sender; 
    if (CallCenterCache::Instance().IsWaiter(sender)) {
      int64_t shop = ImCallCenterAdapter::Instance().GetLinkedShop(sender, receiver, 2);
      LOG_INFO("GetLinkedShop waiter=" << sender << " customer=" << receiver << " shop=" << shop);
      if (shop <= 0) {
        return;
      }
      effective_sender = shop;
    }

    int ret = -1;
    for (int i = 0; i < 2; ++i) {
      ret = redis_->GetImMessages(_return, effective_sender, receiver, offset, limit);
      if (ret >= 0)
        break;
    }
    int size = _return.results.size();
    if (size < limit && ret == 0) {
      ImMessageList cell;
      int32_t cnt = limit - size + 1; //多取一条，检查是否有更多
      mysql_->SelectReadMessages(cell, effective_sender, receiver, offset + size, cnt);
      _return.count = (cell.size() < cnt) ? 0 : -1;//如果取到的数据不够需要取的最大限制，说明没有更多了0表示没有更多

      BOOST_FOREACH(const ImMessage& msg, cell) {
        if (_return.results.size() < limit)
          _return.results.push_back(msg);
      }

      if (_return.count == 0)
        redis_->FlagRead(effective_sender, receiver);
    }
    LOG_INFO("GetMessages sender=" << sender << " effective_sender=" << effective_sender << " receiver" << receiver << " offset=" 
             << offset << " limit=" << limit << " ret_len=" << _return.results.size());
  }

  virtual int32_t GetOfflineMessageCount(const int64_t sender) {
    LOG_INFO("GetOfflineMessageCount called. sender=" << sender);
    int32_t count;
    redis_->GetOfflineImMessageCount(&count, sender);
    return count;
  }

private:
 /* const int seq_id_step_;
  act::config seq_id_conf_;
  scoped_ptr<atomic_count> index_;*/
  scoped_ptr<MessageRedisAdapter> redis_;
  shared_ptr<MessageMysqlAdapter> mysql_;

};

}

int main(int argc, char **argv) {
  char * conf_file = "../conf/message_history_service.conf";
  char opt;
  while ((opt = getopt(argc, argv, "c:")) != -1) {
    switch (opt) {
    case 'c':
      conf_file = optarg;
      break;
    default:
      std::cerr << "Unknown option " << optopt << std::endl;
      return 1;
    }
  }

  ConfigReader cfg(conf_file);
  if (!cfg) {
    std::cerr << "Config file " << conf_file << " read error!" << std::endl;
    return 1;
  }

  LOG_INIT(cfg.Get("log4cplus", "file"), cfg.Get("log4cplus", "level"));
  int port = cfg.GetWithType<int>("message_history_service", "port", 9091);
  LOG_INFO("listens on port " << port);

  using namespace imserver;
  CallCenterCache::Instance().Reload();
  melon::service::ThriftService<MessageHistoryServiceHandler, MessageHistoryServiceProcessor> service;
  service.StartRegister("/im/message_history_service", "1.0", 0, base::GetLocalIp(), port, 1);
  service.Start(port);
  return 0;
}

